package com.源码分析.processor;

import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import reactor.core.publisher.EmitterProcessor;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxProcessor;

// 能够支持多个subscriber，同时还对每个subscriber支持backpressure。它也可以订阅publisher，然后把数据同步重放。
// 它有一个bufferSize参数，用来在发布数据之后还没有订阅者期间的数据，onNext会一直阻塞直到数据被消费；当第一个订阅者订阅之后，它会接收到buffer里头的数据，而后续的订阅者就只能消费到自他们订阅那个时候起发布的数据。
// 当所有的subscriber都取消订阅之后，该processor会清空buffer，并停止接收新的订阅。
public class EmitterProcessorDemo {

	static final Logger LOGGER = LoggerFactory.getLogger(EmitterProcessorDemo.class);

	@Test
    public void testEmitterProcessor() throws InterruptedException {
        int bufferSize = 3; //小于8的会被重置为8
        FluxProcessor<Integer, Integer> processor = EmitterProcessor.create(bufferSize);
        Flux<Integer> flux1 = processor.map(e -> e);
        Flux<Integer> flux2 = processor.map(e -> e * 10);

        IntStream.rangeClosed(1, 8).forEach(e -> {
            LOGGER.info("emit:{}",e);
            processor.onNext(e); //如果发布的未消费数据超过bufferSize,则会阻塞在这里
        });

        flux1.subscribe(e -> {
            LOGGER.info("flux1 subscriber:{}",e);
        });

        IntStream.rangeClosed(9, 10).forEach(e -> {
            LOGGER.info("emit:{}",e);
            processor.onNext(e);
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e1) {
                e1.printStackTrace();
            }
        });

        //这个是后面添加的订阅,只能消费之后发布的数据
        flux2.subscribe(e -> {
            LOGGER.info("flux2 subscriber:{}", e);
        });

        processor.onNext(11);
        processor.onNext(12);

        processor.onComplete();
        processor.blockLast();
    }

}
